package com.suning.sawp.service.impl.companymem;

import java.util.Properties;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class KafkaProducer {
    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducer.class);
    
    private kafka.javaapi.producer.Producer<String, String> producer;
    
    private String topic;
    
    private Properties props = new Properties();
    
    public KafkaProducer(String kafkaZkServer, String topic) {

        // 此处配置的是kafka的端口
        props.put("metadata.broker.list", kafkaZkServer);
        // 配置value的序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        // 配置key的序列化类
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");

        this.topic = topic;

        producer = new kafka.javaapi.producer.Producer<String, String>(new ProducerConfig(props));
    }

    public void produce(String message,String messageNo) {
        LOGGER.info("KafkaProducer send beg,message:{}, messageNo:{}",message,messageNo);
        producer.send(new KeyedMessage<String, String>(topic, messageNo, message));
        producer.close();
        LOGGER.info("KafkaProducer send end,message:{}, messageNo:{}",message,messageNo);
    }

}
